package com.shengzai.stream

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

import java.time.Duration

object Demo1WorldCount {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()

    conf.setMaster("local[2]")

    conf.setAppName("worldCount")

    val spark = new SparkContext(conf)

    val ssc = new StreamingContext(spark, Durations.seconds(8))


    val RIDS: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.146.136", 8888)

    val worldCount: DStream[(String, Int)] = RIDS.flatMap(
      line => {
        line.split(",")
      }
    ).map(
      line => {
        (line, 1)
      }
    ).reduceByKey(
      (x, y) => x + y
    )

    worldCount.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
